package com.atguigu.flink.datastreamapi.sink;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import com.mysql.cj.jdbc.MysqlXADataSource;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcExactlyOnceOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.function.SerializableSupplier;

import javax.sql.XADataSource;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Optional;

/**
 * Created by Smexy on 2023/4/6
 *
 *  XA: 分布式事务。
 *          多个进程使用同一个事务，写一批数据。
 *
 * 注意事项:
 *   1.是否支持XA，取决于数据库厂商，及提供的驱动。 Mysql8支持。
 *
 *   2.部分数据源，在单个Connection上，对XA的个数有限制的，例如Mysql和PostgreSQL，都只允许创建一个XA。
 *   如果向Mysql和PostgreSQL写，必须设置
 *          JdbcExactlyOnceOptions.builder().withTransactionPerConnection(true)
 *
 *   3.mysql8+,必须为用户授权XA_RECOVER_ADMIN ，才代表支持XA。
 *
 *   4.如果希望保证EOS，必须设置 JdbcExecutionOptions.maxRetries == 0，否则可能导致数据重复。
 *
 *   5.MySQLTransactionRollbackException: Lock wait timeout exceeded; try restarting transaction
 *         要写入的表被锁了。
 *         导致后续的事务，尝试写入该表时，获取锁的时间超过了系统的限制。
 *
 *              SHOW GLOBAL VARIABLES LIKE 'innodb_lock_wait_timeout'
 *                  查看innodb引擎允许的锁表的极限时间，默认50s；
 *
 *   6.在分布式事务中，如果在同一个事务中，多次修改了同样的信息，此时会造成锁表。
 *              默认Mysql在Innodb引擎是行锁。
 *              事务1:
 *                  replace into (1,xx,xx)
 *                  replace into (2,xx,xx)
 *                  以上操作，写不同行的数据，锁不同行，相互不干扰。
 *
 *              如果在一个事务中，修改了多次相同的行，此时会造成锁表。
 *               事务2:
 *  *                  replace into (1,xx,a)
 *  *                  replace into (1,xx,b)
 *
 */
public class Demo5_JDBCXASink
{
    public static void main(String[] args) {

       StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //必须开启ck
        env.enableCheckpointing(5000);
        //出错时，不要重启。不加，默认出错就重启
        env.setRestartStrategy(RestartStrategies.noRestart());

        env.setParallelism(2);

        /*
            exactlyOnceSink(
                String sql,
                JdbcStatementBuilder<T> statementBuilder,
                JdbcExecutionOptions executionOptions,
                JdbcExactlyOnceOptions exactlyOnceOptions,  EOS的参数设置
                SerializableSupplier<XADataSource> dataSourceSupplier： 配置数据源
            )
         */
        SinkFunction<WaterSensor> sink = JdbcSink
            .<WaterSensor>exactlyOnceSink(
                "INSERT INTO ws VALUES(?,?,?) " +
                    "ON DUPLICATE KEY UPDATE ts=VALUES(ts), vc = VALUES(vc)",
                new JdbcStatementBuilder<WaterSensor>()
                {
                    @Override
                    public void accept(PreparedStatement ps, WaterSensor waterSensor) throws SQLException {
                        ps.setString(1, waterSensor.getId());
                        ps.setLong(2, waterSensor.getTs());
                        ps.setInt(3, waterSensor.getVc());
                    }
                },
                JdbcExecutionOptions.builder()
                                    //必须设置为0
                                    .withMaxRetries(0)
                                    .withBatchSize(100)
                                    // <= innodb_lock_wait_timeout
                                    .withBatchIntervalMs(500)
                                    .build(),
                JdbcExactlyOnceOptions.builder()
                                      .withTransactionPerConnection(true)
                                      .build(),

                new SerializableSupplier<XADataSource>()
                {
                    //返回一个支持分布式事务的数据源，由厂商提供
                    @Override
                    public XADataSource get() {
                        MysqlXADataSource dataSource = new MysqlXADataSource();
                        dataSource.setUrl("jdbc:mysql://hadoop102:3306/221109?serverTimezone=Asia/Shanghai&useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true");
                        //mysql8+，必须已经授权XA_RECOVER_ADMIN
                        dataSource.setUser("root");
                        dataSource.setPassword("000000");

                        return dataSource;
                    }
                }
            );

        env
                   .socketTextStream("hadoop102", 8888)
                   .map(new WaterSensorMapFunction())
                   .addSink(sink);


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
